Apache Kafka：使用java方式操作消费组和重置分区偏移量（admin api） – 源码巴士

您所在的位置：网站首页 › kafka 生产者命令 › Apache Kafka：使用java方式操作消费组和重置分区偏移量（admin api） – 源码巴士

Apache Kafka：使用java方式操作消费组和重置分区偏移量（admin api） – 源码巴士

2023-04-03 11:38| 来源: 网络整理| 查看: 265

当前版本：kafka_2.12-2.8.0

1. 声明

当前内容主要为本人学习和测试使用java方式操作消费组和重置分区偏移量,主要参考：Apache Kafka官方文档

主要为：

使用java方式实现增删查消费组使用admin api方式重置分区偏移量 2. 准备条件使用命令行方式创建一个带分区的复制系数的topic：test ./bin/kafka-topics.sh --create --topic test --replication-factor 1 --partitions 1 --bootstrap-server 192.168.1.101:9092 查看创建后的topic ./bin/kafka-topics.sh --describe --topic test --bootstrap-server 192.168.1.101:9092

在这里插入图片描述 maven依赖

org.apache.kafka kafka_2.12 2.8.0 org.apache.kafka kafka-clients 2.8.0 3. 使用java方式增删查消费组

消费组的创建(完全是由创建消费者连接当前的kafka的group.id来创建消费组)，相同的消费组共同一个topic中的消息，消费消息后kafka记录偏移量

消费者操作：

public static void main(String[] args) throws Exception { Properties props = new Properties();// 创建配置 // 设置bootstrap.servers并指定当前的kafka的启动的目标地址 props.put(KafkaProperties.bootstrapServersName, "192.168.1.101:9092"); props.put("group.id", "testGroup"); props.put("enable.auto.commit", "true"); props.put("auto.commit.interval.ms", "1000"); props.put("session.timeout.ms", "30000"); // 设置当前的key的解序列化所使用的类 props.put(KafkaProperties.keyDeserializer, KafkaProperties.stringDeserializerName); // 设置当前的value的解序列化所使用的类 props.put(KafkaProperties.valueDeserializer, KafkaProperties.stringDeserializerName); // 创建Kafka的消费者，专门用于消费Kafka中的特定的topic中的消息 @SuppressWarnings("resource") KafkaConsumer consumer = new KafkaConsumer(props); // Kafka Consumer subscribes list of topics here. // 采用订阅的方式消费当前主题的消息 consumer.subscribe(Arrays.asList("test")); // print the topic name System.out.println("Subscribed to topic test" ); while (true) { // 循环的方式拉取数据,其中等待时间为100毫秒 ConsumerRecords records = consumer.poll(100); for (ConsumerRecord record : records) { System.out.println(record); } } }

开启后立刻就关闭，此时这个消费组就会自动创建！

admin api操作：

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException { Properties props = new Properties(); props.put("bootstrap.servers", "192.168.1.101:9092"); /* props.put("client.id", "consumer-testGroup-1"); */ //这里不能有，否则报错 AdminClient client = KafkaAdminClient.create(props); listConsumerGroup(client); listConsumerGroupOffset(client, "testGroup"); // 删除已脱机的消费组 deleteConsumerGroup(client, "testGroup"); client.close(2l, TimeUnit.SECONDS); } /** * * @author hy * @createTime 2021-08-14 09:23:36 * @description 列出当前的消费组 * @param client * */ private static void listConsumerGroup(AdminClient client) { ListConsumerGroupsResult listConsumerGroups = client.listConsumerGroups(); try { Collection collection = listConsumerGroups.all().get(); System.out.println(collection); } catch (InterruptedException | ExecutionException e) { // TODO Auto-generated catch block e.printStackTrace(); } } /** * * @author hy * @createTime 2021-08-14 09:27:27 * @description 通过当前的组id方式获取当前的偏移量 * @param client * @param groupId * */ private static void listConsumerGroupOffset(AdminClient client, String groupId) { ListConsumerGroupOffsetsResult listConsumerGroupOffsets = client.listConsumerGroupOffsets(groupId); try { Map map = listConsumerGroupOffsets.partitionsToOffsetAndMetadata().get(); System.out.println(map); } catch (InterruptedException | ExecutionException e) { // TODO Auto-generated catch block e.printStackTrace(); } } /** * * @author hy * @createTime 2021-08-14 09:30:47 * @description 查看消费组详情 * @param client * @param groupId * */ private static void describeConsumerGroups(AdminClient client, String groupId) { DescribeConsumerGroupsResult describeConsumerGroups = client.describeConsumerGroups(Arrays.asList(groupId)); try { Map map = describeConsumerGroups.all().get(); System.out.println(map.get(groupId)); } catch (InterruptedException | ExecutionException e) { // TODO Auto-generated catch block e.printStackTrace(); } } /** * * @author hy * @createTime 2021-08-14 09:35:59 * @description 删除当前的消费者组 * @param client * @param groupId * */ private static void deleteConsumerGroup(AdminClient client, String groupId) { DeleteConsumerGroupsResult deleteConsumerGroups = client.deleteConsumerGroups(Arrays.asList(groupId)); try { Void void1 = deleteConsumerGroups.all().get(); // 只要不报错就表示删除正常 System.out.println(void1); } catch (InterruptedException | ExecutionException e) { // TODO Auto-generated catch block e.printStackTrace(); } }

执行之后，必须当消费组中没有任何消费者存在的时候才可以删除,否则会报错的！错误为：The group is not empty.

命令行方式

## 查看所有的消费组 ./kafka-consumer-groups.sh --bootstrap-server 192.168.1.101:9092 --list ## 查看某个消费组的详细信息 ./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group ## 查看某个消费组的成员 ./kafka-consumer-groups.sh --bootstrap-server 192.168.1.101:9092 --describe --group my-group --members ## 删除消费组 ./kafka-consumer-groups.sh --bootstrap-server 192.168.1.101:9092 --delete --group my-group --group my-group2 4.使用java方式重置偏移量

注意：重置偏移量的前提！(本人测试后)

所有的消费者都已经关闭(断开kafka进行消费消费) 所有的生产者都已经关闭(不再连接kafka进行生产消息) 重置的分区偏移量时候当前topic必须为XXX-数值的分区(这个一定要存在) admin的api不要带有client.id的属性

上面任何一个存在问题,在重置分区偏移量的时候就会出现一下错误：java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.UnknownMemberIdException: Failed altering consumer group offsets for the following partitions:[XXXX]

在这里插入图片描述

下面是重置偏移量的代码

测试首先使用生产者向test这个topic进行生产消息，并连续发送，最后开启前面的消费者进行消费，此时偏移量达到最大，然后关闭消费者，最后打开消费者发现不再消费(zookeeper记录了偏移量)，使用重置topic的分区偏移量后，再次打开消费者，此时发现消费者继续从0开始消费了

1.第一次打开消费者在这里插入图片描述

2.关闭消费者，再次打开消费者在这里插入图片描述 3.再次关闭消费者，打开重置偏移量(这个地方可能会由于关闭消费者的太快，出现报错，等一下就行了)

这个是重置成功的结果(不报错)

在这里插入图片描述 4.再次打开消费者

在这里插入图片描述再次消费成功！

在测试中最容易出现的错误就是：org.apache.kafka.common.errors.UnknownMemberIdException: Failed altering consumer group offsets for the following partitions: [test-0],注意消费者关闭后等待时间和其他问题，最后的结果是测试成功

【本文地址】

Apache Kafka：使用java方式操作消费组和重置分区偏移量（admin api） – 源码巴士

Apache Kafka：使用java方式操作消费组和重置分区偏移量（admin api） – 源码巴士

今日新闻

推荐新闻